package com.lhz.collector.sink;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.Configurables;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketException;

/**
 * Created by LHZ on 2017/4/17.
 */
public class UDPSink extends AbstractSink implements Configurable {
    private static final Logger logger = LoggerFactory.getLogger(UDPSink.class);
    private String host;
    private int port;
    private InetSocketAddress targetHost;
    private DatagramSocket socket;
    @Override
    public void start() {
        targetHost=new InetSocketAddress(host, port);
        try {
            socket = new DatagramSocket();
        } catch (SocketException e) {
            e.printStackTrace();
        }
        super.start();
    }

    @Override
    public void stop() {
        socket.close();
        super.stop();
    }

    @Override
    public Status process(){
        logger.debug("send event to {}:{}",host,port);
        Status status = Status.READY;
        Channel channel = getChannel();
        Transaction txn = channel.getTransaction();
        try {
            txn.begin();
            Event event = channel.take();
            if(event!=null){
                byte[] bytes=event.getBody();
                socket.send((new DatagramPacket(bytes, 0, bytes.length, targetHost)));
                if(logger.isDebugEnabled()){
                    logger.debug("发送消息：{}",new String(bytes));
                }
            }
            txn.commit();
        } catch (Throwable ex) {
            logger.error("Exception in send event",ex);
            txn.rollback();
        } finally {
            txn.close();
        }
        return status;
    }

    @Override
    public void configure(Context context) {
        Configurables.ensureRequiredNonNull(
                context, "host");
        Configurables.ensureRequiredNonNull(
                context, "port");
        host = context.getString("host");
        port = context.getInteger("port");
    }
}
